package com.zzy.study.doubleThreadNIO;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class Server extends Thread{

	private static final Log LOG = LogFactory.getLog(Server.class);
	private ServerSocketChannel serverChannel = null;
	private Selector selector = null;
	
	private static final int BLOCK_SIZE = 1024;
	private ByteBuffer receiveBuffer = ByteBuffer.allocate(BLOCK_SIZE);
	private ByteBuffer sendBuffer = ByteBuffer.allocate(BLOCK_SIZE);
	private int count;
	private int flag = 0;
	private static final String SEND_STRING = "message from server ";
	
	private BlockingQueue<SocketChannel> acceptQueue = new LinkedBlockingQueue<SocketChannel>();
	private BlockingQueue<SocketChannel> responseQueue = new LinkedBlockingQueue<SocketChannel>();
	
	public Server(int port) {
		init(port);
		this.setName("server");
		new Reader().start();
		new Response().start();
	}

	private void init(int port) {
		try {
			serverChannel = ServerSocketChannel.open();
			selector = Selector.open();
			serverChannel.socket().bind(new InetSocketAddress(port));
			serverChannel.configureBlocking(false);
			serverChannel.register(selector, SelectionKey.OP_ACCEPT);
			LOG.info("server has start, listener port:" + port);
		} catch (IOException e) {
			LOG.error(e);
		}
	}
	
	@Override
	public void run() {
		int count = 0;
		try{
			while(true) {
				count = selector.select();
				if(count == 0) {
					continue;
				}
				
				Set<SelectionKey> selectionKeys = selector.selectedKeys();
				Iterator iterator = selectionKeys.iterator();
				while(iterator.hasNext()) {
					SelectionKey selectionKey = (SelectionKey) iterator.next();
					iterator.remove();
					if(selectionKey.isValid()){
						doAccept(selectionKey);
					}
						
				}
			}
		}catch (Exception e) {
			LOG.error(e);
		}
	}

	private void doAccept(SelectionKey selectionKey) throws IOException {
		if(selectionKey.isAcceptable()) {
			serverChannel = (ServerSocketChannel) selectionKey.channel();
			SocketChannel socketChannel = serverChannel.accept();
			socketChannel.configureBlocking(false);
			socketChannel.register(selector, SelectionKey.OP_READ);
			LOG.info(socketChannel.socket().getRemoteSocketAddress() + " has a request!");
			acceptQueue.add(socketChannel);
			LOG.info(socketChannel + " has add to acceptQueue");
		}
	}
	
	private class Reader extends Thread{
		
		private SocketChannel socketChannel = null;
		
		public Reader() {
			this.setName("reader");
		}
		
		@Override
		public void run() {
			while(true) {
				try {
					socketChannel = acceptQueue.take();
					LOG.info(socketChannel + " has take from acceptQueue");
					receiveBuffer.clear();
					count = socketChannel.read(receiveBuffer);
					if(count > 0) {
						LOG.info(new String(receiveBuffer.array(), 0, count) + " has received from client!");
						socketChannel.register(selector, SelectionKey.OP_WRITE);
						responseQueue.add(socketChannel);
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	
	private class Response extends Thread{
		
		private SocketChannel socketChannel = null;
		
		public Response() {
			this.setName("response");
		}
		
		@Override
		public void run() {
			while(true) {
				try {
					socketChannel = responseQueue.take();
					LOG.info(socketChannel + " has take from responseQueue");
					sendBuffer.clear();  
	    			String msg = SEND_STRING + flag++;
	    			
	    			ByteArrayOutputStream response = new ByteArrayOutputStream(10240);
	    			DataOutputStream out = new DataOutputStream(response);
	    			out.writeInt(msg.length());
	    			out.write(msg.getBytes());
	    			
	    			sendBuffer.put(ByteBuffer.wrap(response.toByteArray()));
	    			sendBuffer.flip();  
	    			socketChannel.write(sendBuffer);  
	    			LOG.info(new String(msg + " has set to client!"));
	    			socketChannel.register(selector, SelectionKey.OP_READ);  
	    			acceptQueue.add(socketChannel);
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	
	public static void main(String[] args) {
		new Server(9111).start();
	}
	
}
